package com.shujia.mr.wordcount;


import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;


/**
 * 自定义的Mapper类需要继承mapreduce包中的Mapper来实现其map函数
 *      需要定义其Mapper类的泛型
 *      <KEYIN, VALUEIN, KEYOUT, VALUEOUT> 该4种数据类型需要使用Hadoop中的序列化类型 Writable
 *          其中KEYIN和VALUEIN表示map阶段输入数据的数据类型
 *       KEYIN：表示读取文件时的偏移量，最好使用长整型的Hadoop序列化类 LongWritable
 *       VALUEIN：表示为读取文件的一行数据，那么对应为字符串的Hadoop序列化类 Text
 *          其中KEYOUT和VALUEOUT表示经过map阶段处理后，输出数据的类型
 *       KEYOUT：表示经过map阶段处理后输出数据的Key类型，在WordCount计算中输出的Key为单词所以对应类型为 Text
 *       VALUEOUT: 示经过map阶段处理后输出数据的Value类型,在WordCount计算中输出的Value是数量1 对应类型为
 */
public class MyWordCountMapper extends Mapper<LongWritable, Text,Text, IntWritable> {

    /**
     * 该map方法是Map阶段用户定义的数据处理逻辑,执行过程中是按行读取数据，做数据处理的
     *      对于wordcount来说，在map方法中，
     *          需要将读取到的一行数据（hello hadoop）进行对单词做切分，切分完成后
     *          再对数据进行处理，添加单词的频数 1 形成 K V 数据格式进行输出
     * @param key 读取数据的偏移量
     * @param value 读取的一行数据
     * @param context MapReduce中上下文对象，将Map阶段和Reduce阶段串起来，同时可以通过该对象进行发送数据
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        // 将读取到的一行数据进行转换成Java的String类型，然后进行单词切分
        String[] words = value.toString().split(" ");
        for (String word : words) {

            // 通过context对象中的write函数将单词和词频写出到Reduce阶段
            // public void write(KEYOUT key, VALUEOUT value)  要求传入KEYOUT和VALUEOUT
            // 根据业务逻辑需要将 word,1 的K/V类型写出，但是需要做类型转换
            context.write(new Text(word),new IntWritable(1));
        }

    }
}
